package com.yangchao.webflux.projectreactor;

import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import reactor.core.Exceptions;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

import java.util.List;

public class SubscribeTest {


    static Logger logger = org.slf4j.LoggerFactory.getLogger(MonoTest.class);

    @Test
    @DisplayName("订阅者")
    public void test01() {
        Flux<Integer> ints = Flux.range(1, 4)
                .map(i -> {
                    if (i <= 3) return i;
                    throw new RuntimeException("Got to 4");
                });
        ints.subscribe(i -> logger.info("value: {} ", i),
                error -> logger.info("Error: {} ,{}", error.getMessage(), error.getStackTrace()
                ));
    }

    @Test
    @DisplayName("订阅者 2")
    public void test02() {
        Flux<Integer> ints = Flux.range(1, 4)
                .map(i -> {
                    if (i <= 3) return i;
                    throw new RuntimeException("Got to 4");
                }).doOnCancel(() -> logger.info("Cancelled"));
        ints.subscribe(i -> logger.info("value: {} ", i),
                error -> logger.info("Error: {} ,{}", error.getMessage(), error.getStackTrace()
                ),
                () -> logger.info("Complete"),
                subscriptionConsumer -> {
                    subscriptionConsumer.request(2);
                    subscriptionConsumer.cancel();
                });
    }

    @Test
    @DisplayName("订阅者 3")
    public void test03() {
        Flux<Integer> ints = Flux.range(1, 6)
                .map(i -> {
                    if (i <= 3) return i;
                    throw new RuntimeException("Got to " + i);
                })
                .onErrorComplete((Throwable throwable) -> {
                    logger.info(" onErrorComplete Error: {} ,{}", throwable.getMessage(), throwable.getStackTrace());
                    return false;
                })
                .doOnError((Throwable throwable) -> logger.info("Error publisher: {} ,{}", throwable.getMessage(), throwable.getStackTrace()));

        ints.subscribe(new BaseSubscriber<Integer>() {

            @Override
            protected void hookOnNext(Integer value) {
                logger.info("value: {} ", value);
                if(value.equals(2)){
                    cancel(); //取消流
                }
                this.requestUnbounded();
            }

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(1);
            }

            @Override
            protected void hookOnError(Throwable throwable) {
                logger.info("Error: {} ,{}", throwable.getMessage(), throwable.getStackTrace());
                // super.hookOnError(throwable);
            }

        });
    }

    @Test
    @DisplayName("订阅者 4")
    public void test04() {
        Flux<List<Integer>> flux = Flux.range(1, 1000)  //原始流10个
                .buffer(3)
                .log();//缓冲区：缓冲3个元素: 消费一次最多可以拿到三个元素； 凑满数批量发给消费者
            //
            //        //一次发一个，一个一个发；
            // 10元素，buffer(3)；消费者请求4次，数据消费完成
        flux.subscribe(new BaseSubscriber<List<Integer>>() {

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(20);
            }


            @Override
            protected void hookOnNext(List<Integer> value) {
                logger.info("value: {} ", value);
                this.request(1);
            }
        });
    }
    @Test
    @DisplayName("订阅者 5")
    public void test05() {
        Flux<Integer> integerFlux = Flux.range(1, 1000)
                .log()
                //限流触发，看上游是怎么限流获取数据的
                .limitRate(100);//一次预取30个元素； 第一次 request(100)，以后request(75)

        integerFlux.subscribe(new BaseSubscriber<Integer>() {
            @Override
            protected void hookOnNext(Integer value) {
                logger.info("value: {} ", value);
                this.request(20);

            }

            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                subscription.request(20);
            }
        });

    }
}
